package com.kuizii.demo.core.service;

import com.kuizii.base.annotaion.AuditLog;
import com.kuizii.base.api.CustomException;
import com.kuizii.base.common.utils.BeanHelper;
import com.kuizii.base.domain.BaseResponse;
import com.kuizii.demo.api.SysUserService;
import com.kuizii.demo.api.UserService;
import com.kuizii.demo.core.data.service.SysUserDataService;
import com.kuizii.demo.core.domain.entity.user.SysUserEntity;
import com.kuizii.demo.domain.SysUserDto;
import com.kuizii.demo.domain.user.AddUserRequest;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.validation.Valid;
import java.io.IOException;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.Properties;

/**
 * <p>
 * 如果业务微复杂 ,由Service层聚合dataService层以及manager层的数据,提供服务 。小型可不再使用该层
 * <p/>
 *
 * @author nuanyang
 * Email freexiaodong@163.com
 * Created by 2019-06-16 16:32:52
 */

@Api(value = "用户表接口文档")
@RestController
@RequestMapping("/core/sys_user")
public class SysUserServiceImpl implements SysUserService {


    @Autowired(required = false)
    RestHighLevelClient restHighLevelClient;

    @Autowired
    SysUserDataService sysUserDataService;

    @Autowired
    UserService userService;

    @AuditLog(description = "测试")
    @GetMapping("/esQuery")
    public void esQuery(String keyword) {

        try {
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            SearchRequest searchRequest = new SearchRequest("test");
            searchSourceBuilder.query(QueryBuilders.matchQuery("desc", keyword));
            searchSourceBuilder.size(1);
            searchRequest.source(searchSourceBuilder);

            SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

            SearchHits hits = search.getHits();
            for (SearchHit hit : hits) {
                Map<String, Object> highlightFields = hit.getSourceAsMap();
              System.out.println(highlightFields);
            }

            //下面是scroll演示
            //String scrollId = search.getScrollId();
            //
            //Scroll scroll = new Scroll(TimeValue.timeValueSeconds(60));

            //SearchScrollRequest scrollRequest = new SearchScrollRequest();
            //scrollRequest.scroll(scroll);
            //scrollRequest.scrollId(scrollId);
            //SearchResponse searchResponse = restHighLevelClient.scroll(scrollRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            System.out.println(e);
        }
    }

    //    @Autowired
//    KafkaTemplate<Object, Object> kafkaTemplate;
    @AuditLog(description = "测试",spel="#args[0]+'测试'")
    @GetMapping("/test")
    public BaseResponse test() {
        try {
//            kafkaTemplate.send("info","log", UUID.randomUUID().toString());

            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.1.149:9092");
            //判断是否成功，我们指定了“all”将会阻塞消息 0.关闭 1.主broker确认 -1（all）.所在节点都确认
            props.put("acks", "0");
            props.put("retries", 2147483647);
            //延迟1s，1s内数据会缓存进行发送
            props.put("linger.ms", 0);
//        props.put("connections.max.idle.ms", 1);
            props.put("batch.size", 16384);
//        props.put("buffer.memory", 3554432);
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//        return new DefaultKafkaProducerFactory<>(new HashMap<>((Map)props)).createNonTransactionalProducer();
            KafkaProducer<String, String> stringStringKafkaProducer = new KafkaProducer<>(props);
            stringStringKafkaProducer.send(new ProducerRecord<>("test111", "vavavav"));
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return BaseResponse.ok();
    }


    @GetMapping("/testAsync")
    public String testAsync() {

        String s = sysUserDataService.testAsync();

        return "res\r\n";
    }

    @ApiOperation("模拟积分变更")
    @PostMapping("addScore")
    public BaseResponse addScore(String userId, BigDecimal score) throws Exception {
        String s = userService.addScore(userId, score);
        return BaseResponse.ok(s);
    }


    @Override
    @GetMapping("/getUserById")
    public SysUserDto getUserById(String userId) {
        SysUserEntity entity = sysUserDataService.getById(userId);
        SysUserDto result = BeanHelper.copyProperties(entity, SysUserDto::new);
        return result;
    }


    @Override
    @PostMapping("/addUser")
    public void addUser(@Valid AddUserRequest userDto) {
        if (userDto == null) {
            throw new CustomException("传入用户信息不存在");
        }
        SysUserEntity entity = BeanHelper.copyProperties(userDto, SysUserEntity::new);
        entity.setCreateTime(LocalDateTime.now());
        sysUserDataService.save(entity);

    }
    //reg


}
